import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.functions._

// 定义一个名为DeleteStudents的对象
object DeleteStudents {
  def main(args: Array[String]): Unit = {
    // 创建一个SparkSession对象，设置应用名称为"DeleteStudents"，并指定在本地[*]上运行
    val spark = SparkSession.builder.appName("DeleteStudents").master("local[*]").getOrCreate()

    // 读取MySQL数据库中的表"students"数据到DataFrame
    val studentsDF = spark.read
      .format("jdbc")
      .option("url", "jdbc:mysql://localhost:3306/sparksql_practice")  // 数据库连接URL
      .option("dbtable", "students")  // 表名
      .option("user", "root")  // 用户名
      .option("password", "123456")  // 密码
      .option("driver", "com.mysql.jdbc.Driver")  // MySQL驱动程序
      .load()

    // 过滤出年龄不超过30岁的学生记录
    val filteredDF = studentsDF.filter(col("age") <= 30)

    // 删除特定ID的学生记录（例如，删除ID为1的学生记录）
    val idToRemove = 1
    val removedDF = filteredDF.filter(col("student_id") =!= idToRemove)

    // 打印删除后的数据
    removedDF.show()

    // 将删除后的数据重新写入数据库
    removedDF.write
      .format("jdbc")
      .option("url", "jdbc:mysql://localhost:3306/sparksql_practice")  // 数据库连接URL
      .option("dbtable", "students")  // 表名
      .option("user", "root")  // 用户名
      .option("password", "123456")  // 密码
      .option("driver", "com.mysql.jdbc.Driver")  // MySQL驱动程序
      .mode("overwrite")  // 覆盖已存在的表数据
      .save()

    spark.stop()  // 停止SparkSession
  }
}
